package qiyu.live.gift.provider.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.Resource;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.idea.qiyu.live.framework.redis.starter.key.GiftProviderCacheKeyBuilder;
import org.qiyu.live.bank.dto.AccountTradeReqDTO;
import org.qiyu.live.bank.dto.AccountTradeRespDTO;
import org.qiyu.live.bank.interfaces.IQiyuCurrencyAccountRpc;
import org.qiyu.live.common.interfaces.topic.ImCoreServerProviderTopicNames;
import org.qiyu.live.im.constants.AppIdEnums;
import org.qiyu.live.im.dto.ImMsgBody;
import org.qiyu.live.msg.enums.ImBiZiCode;
import org.qiyu.live.web.starter.context.QiyuRequestContext;
import org.qiyu.live.web.starter.error.ErrorAssert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import qiyu.live.framework.mq.starter.properties.RocketMQConsumerProperties;
import qiyu.live.gift.dto.SendGiftMq;
import qiyu.live.gift.provider.constants.SendGiftTypeEnum;
import qiyu.live.im.router.interfaces.IMRouterRpc;
import qiyu.live.living.dto.req.LivingRoomReqDTO;
import qiyu.live.living.dto.resp.LivingRoomRespDTO;
import qiyu.live.living.interfaces.ILivingRoomRpc;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.qiyu.live.web.starter.error.QiYuBaseErrorEnum.SEND_GIT_ERROR;

@Component
public class SendGiftConsumer implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(SendGiftConsumer.class);

    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;
    @Resource
    private IQiyuCurrencyAccountRpc currencyAccountRpc;
    @Resource
    private IMRouterRpc imRouterRpc;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private GiftProviderCacheKeyBuilder cacheKeyBuilder;
    @DubboReference
    private ILivingRoomRpc livingRoomRpc;

    // pk进度条的临界值
    private static final long PK_MAX_NUM = 1000L;
    private static final long PK_MIN_NUM = 0L;


    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        //老版本中会开启，新版本的mq不需要使用到
        mqPushConsumer.setVipChannelEnabled(false);
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + SendGiftConsumer.class.getSimpleName());
        //一次从broker中拉取10条消息到本地内存当中进行消费,一次拉取10条消息
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //监听im发送过来的业务消息topic
        mqPushConsumer.subscribe(ImCoreServerProviderTopicNames.SEND_GIFT_TOPIC, "");

        mqPushConsumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                SendGiftMq sendGiftMq = JSON.parseObject(new String(msg.getBody()), SendGiftMq.class);
                // 整一个消息id
                String giftConsumerKey = cacheKeyBuilder.buildGiftConsumerCacheKey(sendGiftMq.getUuid());
                // 解决mq重复消费问题
                Boolean lockStatus = redisTemplate.opsForValue().setIfAbsent(giftConsumerKey, -1, 5, TimeUnit.MINUTES);
                if (Boolean.FALSE.equals(lockStatus)) {
                    // 曾经消费过--->
                    continue;
                }

                AccountTradeReqDTO accountTradeReqDTO = new AccountTradeReqDTO();
                Integer price = sendGiftMq.getPrice();
                accountTradeReqDTO.setNum(price);
                accountTradeReqDTO.setUserId(QiyuRequestContext.getUserId());

                // 还有一种情况就是我之前打过标记,但是调用服务不可用,导致下次我再到sennx的时候就不能再次消费这个消息了,这种情况
                // 虽然不可避免,但是最终只是系统出现故障了,提示用户一下即可,但是用户的钱并没有扣款
                AccountTradeRespDTO tradeRespDTO = currencyAccountRpc.consumeForGift(accountTradeReqDTO);
                if (tradeRespDTO == null) {
                    LOGGER.error("[SendGiftConsumer] tradeRespDTO is null");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                JSONObject jsonObject = new JSONObject();
                Integer giftType = sendGiftMq.getType();
                Integer roomId = sendGiftMq.getRoomId();
                if (giftType == null || roomId == null) {
                    LOGGER.error("[SendGiftConsumer] giftType or roomId is null");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 成功之后(扣减余额..)才发送礼物
                if (tradeRespDTO.isSuccess()) {
                    // 在发送礼物操作之前,我们都要记录一下礼物序号,防止消息叨叨前端发生乱序,保证MQ消息的顺序
                    String seqNumCacheKey = cacheKeyBuilder.buildGiftSeqNumCacheKey(roomId);
                    Long sendGiftSeqNum = redisTemplate.opsForValue().increment(seqNumCacheKey);
                    // 我们这里要检查一下发送礼物的类型是普通类型直播礼物,还是pk类型的礼物
                    if (giftType.equals(SendGiftTypeEnum.DEFAULT_SEND_GIFT.getCode())) {
                        // 发送礼物成功,则依赖IM发送礼物特效给用户--->普通直播间发送礼物(只有主播自己能看见)
                        jsonObject.put("url", sendGiftMq.getUrl());
                        sendImMsgSingleton(sendGiftMq, ImBiZiCode.SEND_GIFT_SUCCESS_BIZ_CODE.getCode(), jsonObject);
                    } else if (giftType.equals(SendGiftTypeEnum.PK_SEND_GIFT.getCode())) {
                        // 对于pk直播间来说,我们不仅要让礼物全直播课件,而且还要让全直播间看到一致的pk进度条
                        // 对于pk进度条它是由送礼物影响的, 比如pk进度条初始值为1000    主播A(500)   主播B(500)
                        // 当pk进度条到达了边界值比如 0 or 1000 那么就要停止,宣布哪一个主播获胜
                        // 1. 首先我们的知道两个互相pk的主播是谁

                        // 1.1 我们能够查出当前主播的id,那么我们根据roomId查询
                        LivingRoomRespDTO respDTO = livingRoomRpc.queryByRoomId(roomId);
                        long pkUserId = respDTO.getAnchorId();
                        // 1.2 查询出当前连线的userId
                        long pkObjId = livingRoomRpc.queryOnlinePkObjId(roomId);
                        // 2. 保存进度条-->pk进度条是以直播间为维度的,所以我们以roomId作为key缓存到redis中,val为pk进度条
                        String cacheKey = cacheKeyBuilder.buildGiftPkCacheKey(roomId);
                        // 2.1 从缓存中取出pk进度条
                        Long pkNum = (Long) redisTemplate.opsForValue().get(cacheKey);
                        // 2.2 获取到礼物接收方
                        Long receiverId = sendGiftMq.getReceiverId();
                        // 2.3 获取到礼物价值,用于更新pk进度条-->prize获取完了
                        if (pkNum == null || receiverId == null) {
                            LOGGER.error("[SendGiftConsumer] tradeRespDTO is null");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        // 3. 看礼物是给谁送的,更新进度条的值
                        // 如果是送给主播A的就incr,到达边界值1000停止送礼物
                        // 如果是送给主播B的就decr,到达边界值0就停止送礼物
                        // 我们还要做的就是当进度条已经超过边界值了,就停止送礼物了,要有一个保护机制,防止为负数或者超过pkNum
                        if (receiverId.equals(pkUserId)) {
                            // 3.1 送给主播A的,就incr
                            Long increment = redisTemplate.opsForValue().increment(cacheKey, price);
                            if (PK_MAX_NUM <= increment) {
                                // 获胜者为主播A
                                jsonObject.put("winnerId", pkUserId);
                                pkNum = PK_MAX_NUM;
                            } else {
                                pkNum = increment;
                            }
                        } else if (receiverId.equals(pkObjId)) {
                            // 3.2 送给主播B的,就decr
                            Long decrement = redisTemplate.opsForValue().decrement(cacheKey, price);
                            if (PK_MIN_NUM <= decrement) {
                                // 获胜者为主播B
                                jsonObject.put("winnerId", pkObjId);
                                pkNum = PK_MIN_NUM;
                            } else {
                                pkNum = decrement;
                            }
                        }
                        // pk直播间,发送给直播间的所有人--->礼物,pk进度条直播间全部可见
                        jsonObject.put("url", sendGiftMq.getUrl());// pk礼物url
                        jsonObject.put("sendGiftSeqNum", sendGiftSeqNum);// pk发送的礼物序号----> 保证MQ的消息顺序
                        jsonObject.put("pkNum", pkNum);// pk进度条
                        sendImMsgBatch(sendGiftMq, ImBiZiCode.SEND_GIFT_PK_SUCCESS_BIZ_CODE.getCode(), jsonObject);
                    }
                } else {
                    // 利用im推送给用户错误信息
                    jsonObject.put("msg", tradeRespDTO.getMsg());
                    sendImMsgSingleton(sendGiftMq, ImBiZiCode.SEND_GIFT_FAIL_BIZ_CODE.getCode(), jsonObject);
                }
                LOGGER.info("[SendGiftConsumer] msg is {}", msg);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        mqPushConsumer.start();
        LOGGER.info("mq消费者启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
    }

    private void sendImMsgBatch(SendGiftMq sendGiftMq, int bizCode, JSONObject jsonObject) {
        // 批量的查询直播间的所有在线用户,构建好imMsgBodies
        LivingRoomReqDTO livingRoomReqDTO = new LivingRoomReqDTO();
        livingRoomReqDTO.setRoomId(sendGiftMq.getRoomId());
        List<Long> userIdList = livingRoomRpc.queryUserIdsByRoomId(livingRoomReqDTO);
        List<ImMsgBody> imMsgBodies = userIdList.stream().map(userId -> {
            ImMsgBody imMsgBody = new ImMsgBody();
            imMsgBody.setAppId(AppIdEnums.QIYU_LIVE_BIZ.getCode());
            imMsgBody.setRoomId(sendGiftMq.getRoomId());
            imMsgBody.setBiZiCode(bizCode);
            imMsgBody.setUserId(sendGiftMq.getReceiverId());
            imMsgBody.setData(jsonObject.toString());
            return imMsgBody;
        }).collect(Collectors.toList());
        imRouterRpc.batchSendMsg(imMsgBodies);
    }

    private void sendImMsgSingleton(SendGiftMq sendGiftMq, Integer bizCode, JSONObject jsonObject) {
        ImMsgBody imMsgBody = new ImMsgBody();
        imMsgBody.setAppId(AppIdEnums.QIYU_LIVE_BIZ.getCode());
        imMsgBody.setRoomId(sendGiftMq.getRoomId());
        imMsgBody.setBiZiCode(bizCode);
        imMsgBody.setUserId(sendGiftMq.getReceiverId());
        imMsgBody.setData(jsonObject.toString());
        // 发送IM消息,如果成功发送礼物特效,失败返回给发送礼物的那个人余额不足
        imRouterRpc.sendMsg(imMsgBody);
    }
}